package com.xxx.flink.sink;

import com.xxx.flink.customsource.CustomSource;
import com.xxx.flink.pojo.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Sink到jdbc
 */
public class SinkToJDBC {

    public static void main(String[] args) {
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);

            // 读取数据
    //        DataStreamSource<Event> sourceStream = env.fromElements(
    //                new Event("Bob", "./home", 1690726295934L),
    //                new Event("LiLy", "./home", 1690726354943L),
    //                new Event("Bob", "./cart", 1690726361950L),
    //                new Event("Ketty", "./cart", 1690726366944L),
    //                new Event("Alen", "./home", 1690726372948L),
    //                new Event("Bob", "./product", 1690726377945L)
    //        );
            DataStreamSource<Event> sourceStream = env.addSource(new CustomSource());

            // sink到mysql
            JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                    .withDriverName("com.mysql.cj.jdbc.Driver")
                    .withUrl("jdbc:mysql://localhost:3306/test?connectTimeout=1000&zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=utf-8&useSSL=false&allowMultiQueries=true&rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai")
                    .withUsername("root")
                    .withPassword("123456")
                    .build();
            String sql = "insert into clicks(name, url) values(?, ?)";
            sourceStream.addSink(JdbcSink.sink(
                    sql,
                    ((statement, event) -> {
                        statement.setString(1, event.getName());
                        statement.setString(2, event.getUrl());
                    }),
                    jdbcConnectionOptions));

            env.execute("mysql");
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
